package com.zhaosc.spark.stream

import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.OffsetRange
import org.apache.spark.streaming.kafka.HasOffsetRanges

object SparkStreamingOnKafkaReceiver {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc = new StreamingContext(conf, Seconds(5)) //5秒处理一次消息

    val topic = Map("test" -> 1);
    val kafkaStream = KafkaUtils.createStream(ssc, "localhost:2181", "MyFirstConsumerGroup", topic, StorageLevel.MEMORY_AND_DISK);

    //    val numStreams = 5
    //    val kafkaStreams = (1 to numStreams).map { i => KafkaUtils.createStream(ssc, "localhost:2181", "MyFirstConsumerGroup", topic, StorageLevel.MEMORY_AND_DISK) }
    //    val unifiedStream = ssc.union(kafkaStreams)
    /**
     * kafkaStream是一个kv格式的  k:offset v:values
     */
    val wordsCount = kafkaStream.flatMap(v => {
      v._2.split(" ")
    }).map(v => {
      Tuple2(v, 1)
    }).reduceByKey(_ + _);

    wordsCount.foreachRDD(rdd => {
      rdd.foreach(println(_))
    })
    ssc.start();
    ssc.awaitTermination();

  }
}